﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using AzureMRCore;
using AzureMRCore.DataTypes;
using Microsoft.WindowsAzure;
using System.Collections;
using Microsoft.WindowsAzure.StorageClient;
using System.Threading;
using System.Diagnostics;
using AzureMRCore.DataModel;
using AzureMRCore.Client;
using System.IO;
using System.Threading.Tasks;
using AzureMRCore.IterativeDataCache;

namespace WCClient
{
    class Program
    {

        //static StorageCredentials credentials = new StorageCredentialsAccountAndKey("account","key");
        //    "udh6XItvt9BBHWGIQ7kKe4wppedD6USPd3LXAYimbGZYu8/egbtiB0XsPBpltnPMQYCSQw4ayJJ2m/d7EQlyGg==");
        
        // args : num_seq block_size maxBlocks jobname
        static void Main(string[] args)
        {
            //CloudStorageAccount storageAccount = new CloudStorageAccount(credentials, new Uri("http://swgmr.blob.core.windows.net/"),
            //                        new Uri("http://swgmr.queue.core.windows.net/"), new Uri("http://swgmr.table.core.windows.net/"));
            //CloudStorageAccount storageAccount = CloudStorageAccount.DevelopmentStorageAccount;
            StorageCredentials credentials = new StorageCredentialsAccountAndKey("kmeans", "UxZvQFXZJg7xOxQ3dd8zRMZzbdww5I35GiHVirprR8rzWgFYce/8CvuAIkGns5e9I/e/O0mF+hjULMK/RGpsLA==");

            CloudStorageAccount storageAccount = new CloudStorageAccount(credentials, new Uri("https://kmeans.blob.core.windows.net"), new Uri("https://kmeans.queue.core.windows.net"), new Uri("https://kmeans.table.core.windows.net"));
            
            if (args.Length < 6)
            {
                Console.Out.WriteLine("Usage:  AMRWCClient <jobID> <inputContainerName> <outContainerName> <NumberOfReduceTasks> <Cluster Centers> <vector length>");
            }
            string jobID = args[0];
            string inputContainer= args[1];
            string outputContainer = args[2];
            int numReduceTasks = Int32.Parse(args[3]) ;
            string broadcastURI = args[4];
            string vectorLength = args[5];

            CloudTableClient tableClient = storageAccount.CreateCloudTableClient();

            /*long count = 0;
            for (int i = 0; i < 10; i++)
            {


                Stopwatch sw = Stopwatch.StartNew();
                MapMetaDataContext mapDataContext = new MapMetaDataContext(storageAccount.TableEndpoint.ToString(),
                                                                           storageAccount.Credentials,
                                                                           DataModelUtils.GetMapTableName("tc19"));
                IQueryable<MapDataModel> mapTaskEnum = mapDataContext.MapTasks;
                Console.Out.WriteLine(mapTaskEnum.ToList().Count);

                count += sw.ElapsedMilliseconds;
            }
            Console.Out.WriteLine(count/10);
            

            JobBoardDataContext jobBoardDataContext = new JobBoardDataContext(storageAccount.TableEndpoint.ToString(), storageAccount.Credentials,
                            DataModelUtils.getJobBoardTableName());
                    tableClient.CreateTableIfNotExist(jobBoardDataContext.name);
                    IEnumerator<JobBoardDataModel> jobEnum = jobBoardDataContext.Tasks.GetEnumerator();

                   int count = jobDataContext.Jobs.ToList<JobDataModel>().Count;

                    // for some reason doing a "where" on empty table gives an error
                    if (count != 0)
                    {
                        count = jobDataContext.Jobs.Where(p => p.JobID == jobid)
                            .ToList<JobDataModel>().Count;
                    }*/
           /* DataCache dataCache =  new DataCache(storageAccount);

                    if (jobEnum.MoveNext())
                    {
                        //jobEnum.Current.
                        JobBoardDataModel job = jobEnum.Current;
                        MapMetaDataContext mapDataContext = new MapMetaDataContext(storageAccount.TableEndpoint.ToString(), storageAccount.Credentials,
                           DataModelUtils.getMapTableName(job.JobID));
                        IQueryable<MapDataModel> mapTaskEnum = mapDataContext.MapTasks.Where(p => (p.jobID == job.JobID & p.Iteration == job.Iteration &
                            p.TaskStatus != MapDataModel.EXECUTING & p.TaskStatus != MapDataModel.FINISHED & p.blobURL != dataCache.inMemCache.Keys));

                        List<MapDataModel> list = new List<MapDataModel>();
                        IQueryable<MapDataModel> intersect = mapTaskEnum.Intersect(list);
                        //var selectedOrders = from o in mapDataContext.MapTasks
                        //                     where o.jobID == job.JobID                                             
                        //                     select o;

                        //IQueryable<string> blobList = mapTaskEnum.Select(p => p.blobURL);
                        //IEnumerable en =  mapTaskEnum.AsEnumerable();
                        foreach (var blob in intersect)
                        {
                            //Console.Out.WriteLine(blob);
                        }
                        /*IQueryable<string> intersect =  blobList.Intersect(dataCache.inMemCache.Keys);

                        if (intersect.Count() > 0)
                        {
                            string cachedBlob = intersect.First();
                           MapDataModel mapTask =  mapTaskEnum.Where(p => p.blobURL == cachedBlob).First();

                        }
                        //dataCache.
                        //find tasks in that map table matched the data cache
                        //else try the queue
                    }*/
            Stopwatch stopwatch = Stopwatch.StartNew();
            ClientUtils client = new ClientUtils(storageAccount, jobID, "mapsched4", "reducesched4");
            client.ProcessMapRed(inputContainer, 0, vectorLength, numReduceTasks, outputContainer,broadcastURI,true);
            client.WaitForCompletion(500);
            Console.Out.WriteLine("Time (ms) "+stopwatch.ElapsedMilliseconds);

            /*CloudBlobClient blobClient = storageAccount.CreateCloudBlobClient();
            CloudBlobContainer blobContainer = blobClient.GetContainerReference("test256");
            foreach (var blob in blobContainer.ListBlobs())
            {
                CloudBlockBlob oriBlob = blobContainer.GetBlockBlobReference(blob.Uri.ToString());
                CloudBlockBlob copyBlob = blobContainer.GetBlockBlobReference(blob.Uri+"c");
                copyBlob.CopyFromBlob(oriBlob);

                CloudBlockBlob copyBlob2 = blobContainer.GetBlockBlobReference(blob.Uri + "b");
                copyBlob2.CopyFromBlob(oriBlob);

                CloudBlockBlob copyBlob3 = blobContainer.GetBlockBlobReference(blob.Uri + "d");
                copyBlob3.CopyFromBlob(oriBlob);
            }
            //ClientUtils.RemoveTaskTables("t",storageAccount);

           
            /*MapMetaDataContext mapDataContext = new MapMetaDataContext(storageAccount.TableEndpoint.ToString(),
                storageAccount.Credentials, DataModelUtils.getMapTableName(jobID));
            ClientUtils.AddBLOBMapsWithNoQueing(inputContainer, jobID, "", numReduceTasks, outputContainer, 1, storageAccount,mapDataContext);
            JobBoardDataContext jobBoardDataContext = new JobBoardDataContext(storageAccount.TableEndpoint.ToString(), storageAccount.Credentials,
                            DataModelUtils.getJobBoardTableName());
            //CloudTableClient tableClient = storageAccount.CreateCloudTableClient();
            tableClient.CreateTableIfNotExist(jobBoardDataContext.name);

            JobBoardDataModel jobIteration = new JobBoardDataModel(jobID,mapDataContext.name,1);
            jobBoardDataContext.AddObject(jobBoardDataContext.name, jobIteration);
            jobBoardDataContext.SaveChangesWithRetries();

            ClientUtils.WaitForCompletion(jobID, storageAccount, 500);

            ClientUtils.AddBLOBMapsWithNoQueing(inputContainer, jobID, "", numReduceTasks, outputContainer, 2, storageAccount, mapDataContext);
            jobIteration.Iteration = 2;
            jobBoardDataContext.UpdateObject(jobIteration);
            jobBoardDataContext.SaveChangesWithRetries();
            ClientUtils.WaitForCompletion(jobID, storageAccount, 500);

            ClientUtils.AddBLOBMapsWithNoQueing(inputContainer, jobID, "", numReduceTasks, outputContainer, 3, storageAccount, mapDataContext);
            jobIteration.Iteration = 3;
            jobBoardDataContext.UpdateObject(jobIteration);
            jobBoardDataContext.SaveChangesWithRetries();
            //AzureMRClient.RemoveTaskTables("jobwcone",storageAccount);
                */
            Console.In.ReadLine();           
        }    
    }
}
